package net.sh.dpinfrastructure.search;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.action.DocWriteResponse.Result;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequestBuilder;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateRequestBuilder;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.sort.SortOrder;

import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Maps;

import net.sh.dpinfrastructure.db.DPCriteriaType;
import net.sh.dpinfrastructure.db.DPOrderType;
import net.sh.dpinfrastructure.db.DPPageAndSort;
import net.sh.dpinfrastructure.db.DPSearchStatement;
import net.sh.dpinfrastructure.db.DPSystemFields;

/**
 * 
 * {@link DPSearch}的包装elasticsearch的实现类来简化搜索操作
 * @author Lancer
 * @date 2016年4月16日下午5:12:23
 *
 */
public class DPElasticsearch implements DPSearch {
    
    /**
     * 批量删除时每次删除的块大小
     */
    public static final int BATCH_DELETE_PER_SIZE = 100;
    
    /**
     * 默认返回记录条数
     */
    public static final int DEFAULT_LIMIT = 10;

    /**
     * 滚动操作时的超时时间
     */
    public static final int SCROLL_TIME_OUT = 60000;

    private TransportClient client;
    
    /**
     * 包装 {@link TransportClient}
     * @param client 被包装的 {@link TransportClient}实例不能为空
     */
    public DPElasticsearch(TransportClient client) {
        if (client == null) {
            throw new NullPointerException("client");
        }
        this.client = client;
    }
    
    /**
     * 
     * 根据指定的搜索条件构造elasticsearch搜索语句
     *
     * @param criteria
     *
     * @author Lancer
     * @date 2016年5月7日下午6:29:23
     *
     */
    @SuppressWarnings("rawtypes")
    public QueryBuilder buildElasticsearchQuery(DPSearchStatement criteria) {
        LinkedHashMap<DPCriteriaType, Map<String, ? extends Object>> criteriaMap = criteria.toMap();
        if (criteria == null || criteriaMap.isEmpty()) {
            return QueryBuilders.matchAllQuery();
        }
        BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery();
        for (Map.Entry<DPCriteriaType, Map<String, ? extends Object>> criteriaItem : criteriaMap.entrySet()) {
            DPCriteriaType criteriaType = criteriaItem.getKey();
            Map.Entry<String, ? extends Object> fieldValueEntry = criteriaItem.getValue().entrySet().iterator().next();
            String field = fieldValueEntry.getKey();
            Object value  = fieldValueEntry.getValue();
            switch (criteriaType) {
                case EQ: {
                    queryBuilder.must(QueryBuilders.termQuery(field, value));
                    break;
                }
                case NEQ: {
                    queryBuilder.must(QueryBuilders.boolQuery().mustNot(QueryBuilders.termQuery(field, value)));
                    break;
                }
                case LIKE: {
                    String text = (String) value; 
                    queryBuilder.must(QueryBuilders.queryStringQuery(text).field(field));
                    break;
                }
                case UNLIKE: {
                    String text = (String) value; 
                    queryBuilder.mustNot(QueryBuilders.queryStringQuery(text).field(field));
                    break;
                }
                case GT: {
                    queryBuilder.must(QueryBuilders.rangeQuery(field).gt(value));
                    break;
                }
                case GTE: {
                    queryBuilder.must(QueryBuilders.rangeQuery(field).gte(value));
                    break;
                }
                case LT: {
                    queryBuilder.must(QueryBuilders.rangeQuery(field).lt(value));
                    break;
                }
                case LTE: {
                    queryBuilder.must(QueryBuilders.rangeQuery(field).lte(value));
                    break;
                }
                case IN: {
                    QueryBuilder inQuery = null;
                    if (value instanceof Object[]) {
                        inQuery = QueryBuilders.termsQuery(field, (Object[]) value);
                    } else if (value instanceof List) {
                        inQuery = QueryBuilders.termsQuery(field, (List) value);
                    }
                    queryBuilder.must(inQuery);
                    break;
                }
                case NIN: {
                    QueryBuilder ninQuery = null;
                    if (value instanceof Object[]) {
                        ninQuery = QueryBuilders.termsQuery(field, (Object[]) value);
                    } else if (value instanceof List) {
                        ninQuery = QueryBuilders.termsQuery(field, (List) value);
                    }
                    queryBuilder.must(QueryBuilders.boolQuery().mustNot(ninQuery));
                    break;
                }
                case BETWEEN: {
                    List betweenValues = (List) value;
                    Object smallerValue = betweenValues.get(0);
                    Object biggerValue = betweenValues.get(1);
                    queryBuilder.must(QueryBuilders.rangeQuery(field).gt(smallerValue).lt(biggerValue));
                    break;
                }
                case LBETWEEN: {
                    List betweenValues = (List) value;
                    Object smallerValue = betweenValues.get(0);
                    Object biggerValue = betweenValues.get(1);
                    queryBuilder.must(QueryBuilders.rangeQuery(field).gte(smallerValue).lt(biggerValue));
                    break;
                }
                case RBETWEEN: {
                    List betweenValues = (List) value;
                    Object smallerValue = betweenValues.get(0);
                    Object biggerValue = betweenValues.get(1);
                    queryBuilder.must(QueryBuilders.rangeQuery(field).gt(smallerValue).lte(biggerValue));
                    break;
                }
                case LRBETWEEN: {
                    List betweenValues = (List) value;
                    Object smallerValue = betweenValues.get(0);
                    Object biggerValue = betweenValues.get(1);
                    queryBuilder.must(QueryBuilders.rangeQuery(field).gte(smallerValue).lte(biggerValue));
                    break;
                }
                case AND: {
                    DPSearchStatement andCriteria = (DPSearchStatement) value;
                    queryBuilder.must(buildElasticsearchQuery(andCriteria));
                    break;
                }
                case OR: {
                    DPSearchStatement orCriteria = (DPSearchStatement) value;
                    queryBuilder.should(buildElasticsearchQuery(orCriteria));
                    break;
                }
                case NOT: {
                    DPSearchStatement notCriteria = (DPSearchStatement) value;
                    queryBuilder.mustNot(buildElasticsearchQuery(notCriteria));
                    break;
                }
            }
        }
        return queryBuilder;
    }

    /**
     * 
     * 检查库名和表名
     *
     * @param database
     * @param table
     * 
     * @author shaohui
     * @date 2017年2月13日 下午4:44:50
     */
    public void checkDBAndTable(String table) {
        
        if (StringUtils.isEmpty(table)) {
            throw new IllegalArgumentException("empty table name");
        }
    }
    
    @Override
    public long count(String table) {
        return count(table, null);
    }

    @Override
    public long count(String table, DPSearchStatement criteria) {
        checkDBAndTable(table);
        
        SearchRequestBuilder requestBuilder = client.prepareSearch(table).setTypes(table).setSize(0)
                .setSearchType(SearchType.QUERY_THEN_FETCH);
        if (criteria != null) {
            requestBuilder.setQuery(buildElasticsearchQuery(criteria));
        }
        return requestBuilder.execute().actionGet().getHits().totalHits();
    }
    
    @Override
    public int delete(String table, DPSearchStatement criteria) {
        return delete(table, criteria, true);
    }

    @Override
    public int delete(String table, DPSearchStatement criteria, boolean logically) {
        checkDBAndTable(table);
        
        if (logically) {
            return update(table, criteria, ImmutableMap.of(DPSystemFields.STATUS, DPSystemFields.DELETE_FLAG));
        }
        SearchRequestBuilder requestBuilder = client.prepareSearch(table).setTypes(table)
                .setScroll(new TimeValue(SCROLL_TIME_OUT));
        if (criteria != null) {
            requestBuilder.setQuery(buildElasticsearchQuery(criteria));
        }
        SearchResponse scrollResp = requestBuilder.setSize(BATCH_DELETE_PER_SIZE).execute().actionGet();
        int effectedCount = 0;
        while (true) {
            SearchHit[] searchHits = scrollResp.getHits().getHits();
            if (ArrayUtils.isEmpty(searchHits)) {
                break;
            }
            BulkRequestBuilder bulkRequestBuilder = client.prepareBulk();
            for (SearchHit searchHit : searchHits) {
                bulkRequestBuilder.add(new DeleteRequest(table).type(table).id(searchHit.getId()));
            }
            BulkItemResponse[] responses = bulkRequestBuilder.execute().actionGet().getItems();
            for (BulkItemResponse item : responses) {
                effectedCount += (item.<DeleteResponse>getResponse().getResult() == Result.DELETED) ? 1 : 0;
            }
            String scrollId = scrollResp.getScrollId();
            scrollResp = client.prepareSearchScroll(scrollId).setScroll(new TimeValue(SCROLL_TIME_OUT)).execute().actionGet();
        }
        return effectedCount;
    }
    
    @Override
    public int delete(String table, long id) {
        return client.prepareDelete(table, table, String.valueOf(id)).get().getResult() == Result.DELETED ? 1 : 0;
    }

    @Override
    public int delete(String table, String id) {
        return client.prepareDelete(table, table, id).get().getResult() == Result.DELETED ? 1 : 0;
    }

    public TransportClient getClient() {
        return client;
    }

    @Override
    public int insert(String table, List<Map<String, Object>> sources) {
        checkDBAndTable(table);
        
        int affectedNumber = 0;
        BulkRequestBuilder bulkRequestBuilder = client.prepareBulk();
        for (Map<String, Object> source : sources) {
            Map<String, Object> sourceWithStatus = Maps.newHashMap(source);
            sourceWithStatus.put(DPSystemFields.STATUS, DPSystemFields.EXIST_FLAG);
            IndexRequest indexRequest = new IndexRequest(table, table);
            Object id = sourceWithStatus.get("_id");
            if (id != null) {
                indexRequest.id(id.toString());
            }
            sourceWithStatus.remove("_id");
            indexRequest.source(sourceWithStatus);
            sourceWithStatus.put("_id", id);
            bulkRequestBuilder.add(indexRequest);
        }
        BulkItemResponse[] responses = bulkRequestBuilder.execute().actionGet().getItems();
        for (BulkItemResponse response : responses) {
            IndexResponse indexResponse = response.<IndexResponse>getResponse();
            if (indexResponse == null) {
                continue;
            }
            affectedNumber += (indexResponse.getResult() == Result.CREATED) ? 1 : 0;
        }
        return affectedNumber;
    }

    @Override
    public int insert(String table, Map<String, Object> source) {
        
        Map<String, Object> sourcesWithStatus = Maps.newHashMap(source);
        sourcesWithStatus.put(DPSystemFields.STATUS, DPSystemFields.EXIST_FLAG);
        
        Object id = sourcesWithStatus.get("_id");
        int affectedNumber = 0;
        if (id != null) {
            sourcesWithStatus.remove("_id");
            affectedNumber = client.prepareIndex(table, table, id.toString()).setSource(sourcesWithStatus).get().getResult() == Result.CREATED ? 1 : 0;
        } else {
            affectedNumber = client.prepareIndex(table, table).setSource(sourcesWithStatus).get().getResult() == Result.CREATED ? 1 : 0;;
        }
        return affectedNumber;
    }

    @Override
    public List<Map<String, Object>> query(String table) {
        return query(table, null, null, (Map<String, String>) null, null);
    }

    @Override
    public List<Map<String, Object>> query(String table, DPSearchStatement criteria) {
        return query(table, criteria, null, (Map<String, String>) null, null);
    }

    @Override
    public <T> List<T> query(String table, DPSearchStatement criteria,
            DPDocumentMapper<T> mapper) {
        return query(table, criteria, null, mapper, null);
    }
    
    @Override
    public <T> List<T> query(String table, DPSearchStatement criteria,
            DPDocumentMapper<T> mapper, DPPageAndSort pageAndSort) {
        return query(table, criteria, null, mapper, pageAndSort);
    }

    @Override
    public List<Map<String, Object>> query(String table, DPSearchStatement criteria,
            DPPageAndSort extendCondition) {
        return query(table, criteria, null, (Map<String, String>) null, extendCondition);
    }

    @Override
    public List<Map<String, Object>> query(String table, DPSearchStatement criteria,
            Map<String, String> fieldsMap) {
        return query(table, criteria, null, fieldsMap, null);
    }

    @Override
    public List<Map<String, Object>> query(String table, DPSearchStatement criteria,
            Map<String, String> fieldsMap, DPPageAndSort pageAndSort) {
        return query(table, criteria, null, fieldsMap, pageAndSort);
    }

    @Override
    public List<Map<String, Object>> query(String table, DPSearchStatement criteria,
            Set<String> fields) {
        return query(table, null, fields, (Map<String, String>) null, null);
    }

    @Override
    public <T> List<T> query(String table, DPSearchStatement criteria, Set<String> fields,
            DPDocumentMapper<T> mapper) {
        return query(table, criteria, fields, mapper, null);
    }

    @Override
    public <T> List<T> query(String table, DPSearchStatement criteria, Set<String> fields,
            DPDocumentMapper<T> mapper, DPPageAndSort pageAndSort) {
        List<Map<String, Object>> sourceMap = query(table, criteria, fields, (Map<String, String>) null, pageAndSort);
        List<T> resultList = new ArrayList<T>();
        for (Map<String, Object> source : sourceMap) {
            resultList.add(mapper.mapper(source));
        }
        return resultList;
    }

    @Override
    public List<Map<String, Object>> query(String table, DPSearchStatement criteria,
            Set<String> fields, DPPageAndSort pageAndSort) {
        return query(table, criteria, fields, (Map<String, String>) null, pageAndSort);
    }

    @Override
    public List<Map<String, Object>> query(String table, DPSearchStatement criteria,
            Set<String> fields, Map<String, String> fieldsMap) {
        return query(table, criteria, fields, fieldsMap, null);
    }

    
    @Override
    public List<Map<String, Object>> query(String table, DPSearchStatement criteria,
            Set<String> fields, Map<String, String> fieldsMap, DPPageAndSort pageAndSort) {
        checkDBAndTable(table);
        
        SearchRequestBuilder requestBuilder = client.prepareSearch(table).setTypes(table);
        if (criteria != null) { // 查询条件不为空，则应用查询条件
            if (criteria.isFullTextSearch()) { // 全文检索
                requestBuilder.setQuery(buildElasticsearchQuery(criteria));
            } else { // 普通查询
                requestBuilder.setPostFilter(buildElasticsearchQuery(criteria));
            }
        }
        boolean containsId = true; // 是否返回id
        if (fields != null && !fields.isEmpty()) { // 返回字段不为空则应用返回字段
            requestBuilder.storedFields(fields.toArray(new String[] {}));
            if (!fields.contains("_id")) {
                containsId = false;
            }
        }
        int page = 1;
        int rows = DEFAULT_LIMIT;
        String orderField = null;
        LinkedHashMap<String, DPOrderType> orderFields = null; // 复合排序
        if (pageAndSort != null) {
            page = pageAndSort.getPage();
            rows = pageAndSort.getRows();
            orderField = pageAndSort.getOrderField();
            orderFields = pageAndSort.getOrders();
        }
        int skip = (page - 1) * rows;
        requestBuilder.setFrom(skip).setSize(rows); // 设置分页
        
        // 设置排序
        if (StringUtils.isNotBlank(orderField)) { // 排序字段不为空则应用排序
            requestBuilder.addSort(orderField, pageAndSort.getOrderType() == DPOrderType.ASC  ? SortOrder.ASC : SortOrder.DESC);
        }
        if (MapUtils.isNotEmpty(orderFields)) {
            for (Map.Entry<String, DPOrderType> entry : orderFields.entrySet()) {
                String field = entry.getKey();
                if (StringUtils.isEmpty(field) || entry.getValue() == null) {
                    continue;
                }
                SortOrder order = (entry.getValue() == null || entry.getValue() == DPOrderType.ASC ) ? SortOrder.ASC : SortOrder.DESC;
                requestBuilder.addSort(field, order);
            }
        }
        
        //执行查询
        Iterator<SearchHit> iterator = requestBuilder.execute().actionGet().getHits().iterator();
        List<Map<String, Object>> resultList = new ArrayList<Map<String, Object>>();
        if (!MapUtils.isEmpty(fieldsMap)) {
            while (iterator.hasNext()) {
                SearchHit hit = iterator.next();
                if (!hit.hasSource()) {
                    continue;
                }
                Map<String, Object> source = hit.getSource();
                if (containsId) {
                    source.put("_id", hit.getId());
                }
                Map<String, Object> fieldChangedSource = new HashMap<String, Object>();
                for (Map.Entry<String, Object> entry : source.entrySet()) {
                    String originalKey = entry.getKey();
                    String mappedKey = fieldsMap.get(originalKey);
                    if (StringUtils.isEmpty(mappedKey)) {
                        fieldChangedSource.put(originalKey, entry.getValue());
                    } else {
                        fieldChangedSource.put(mappedKey, entry.getValue());
                    }
                }
                resultList.add(source);
            }
        } else {
            while (iterator.hasNext()) {
                SearchHit hit = iterator.next();
                Map<String, Object> source = hit.getSource();
                if (containsId) {
                    source.put("_id", hit.getId());
                }
                resultList.add(source);
            }
        }
        
        return resultList;
    }

    @Override
    public List<Map<String, Object>> query(String table, Set<String> fields) {
        return query(table, null, fields, (Map<String, String>) null, null);
    }

    @Override
    public <T> List<T> query(String table, Set<String> fields, DPDocumentMapper<T> mapper) {
        return query(table, null, fields, mapper, null);
    }

    @Override
    public <T> List<T> query(String table, Set<String> fields, DPDocumentMapper<T> mapper,
            DPPageAndSort extendCondition) {
        return query(table, null, fields, mapper, extendCondition);
    }

    @Override
    public List<Map<String, Object>> query(String table, Set<String> fields,
            DPPageAndSort extendCondition) {
        return query(table, (DPSearchStatement) null, fields, (Map<String, String>) null, extendCondition);
    }

    @Override
    public List<Map<String, Object>> query(String table, Set<String> fields,
            Map<String, String> fieldsMap) {
        return query(table, null, fields, fieldsMap, null);
    }

    @Override
    public List<Map<String, Object>> query(String table, Set<String> fields,
            Map<String, String> fieldsMap, DPPageAndSort extendCondition) {
        return query(table, null, fields, fieldsMap, extendCondition);
    }

    @Override
    public Map<String, Object> queryForMap(String table, DPSearchStatement criteria) {
        return queryForMap(table, criteria, null, null);
    }

    @Override
    public Map<String, Object> queryForMap(String table, DPSearchStatement criteria,
            Map<String, String> fieldsMap) {
        return queryForMap(table, criteria, null, fieldsMap);
    }

    @Override
    public Map<String, Object> queryForMap(String table, DPSearchStatement criteria,
            Set<String> fields) {
        return queryForMap(table, criteria, fields, null);
    }

    @Override
    public Map<String, Object> queryForMap(String table, DPSearchStatement criteria,
            Set<String> fields, Map<String, String> fieldsMap) {
        DPPageAndSort condition = new DPPageAndSort();
        condition.setRows(1);
        List<Map<String, Object>> searchResults = query(table, criteria, fields, fieldsMap, condition);
        if (searchResults.isEmpty()) {
            return null;
        }
        return searchResults.get(0);
    }

    @Override
    public Map<String, Object> queryForMap(String table, long id) {
        return queryForMap(table, String.valueOf(id), null, null);
    }

    @Override
    public Map<String, Object> queryForMap(String table, long id,
            Map<String, String> fieldsMap) {
        return queryForMap(table, String.valueOf(id), null, fieldsMap);
    }

    @Override
    public Map<String, Object> queryForMap(String table, long id, Set<String> fields) {
        return queryForMap(table, String.valueOf(id), fields, null);
    }

    @Override
    public Map<String, Object> queryForMap(String table, long id, Set<String> fields,
            Map<String, String> fieldsMap) {
        return queryForMap(table, String.valueOf(id), fields, fieldsMap);
    }

    @Override
    public Map<String, Object> queryForMap(String table, String id) {
        return queryForMap(table, id, null, null);
    }

    @Override
    public Map<String, Object> queryForMap(String table, String id,
            Map<String, String> fieldsMap) {
        return queryForMap(table, id, null, fieldsMap);
    }

    @Override
    public Map<String, Object> queryForMap(String table, String id, Set<String> fields) {
        return queryForMap(table, id, fields, null);
    }

    @Override
    public Map<String, Object> queryForMap(String table, String id, Set<String> fields,
            Map<String, String> fieldsMap) {
        GetRequestBuilder requestBuilder = client.prepareGet(table, table, id);
        boolean containsId = true; //是否返回_id字段
        if (!CollectionUtils.isEmpty(fields)) {
            requestBuilder.setStoredFields(fields.toArray(new String[] {}));
            if (!fields.contains("_id")) {
                containsId = false;
            }
        }
        GetResponse response = requestBuilder.execute().actionGet();
        Map<String, Object> source = response.getSource();
        if (source != null && containsId) {
            source.put("_id", response.getId());
        }
        if (MapUtils.isEmpty(fieldsMap)) {
            return source;
        }
        Map<String, Object> mappedSource = new HashMap<String, Object>();
        for (Map.Entry<String, Object> entry : source.entrySet()) {
            String originalKey = entry.getKey();
            String mappedKey = fieldsMap.get(originalKey);
            mappedSource.put(StringUtils.isEmpty(mappedKey) ? originalKey : mappedKey, entry.getValue());
        }
        return mappedSource;
    }

    @Override
    public <T> T queryForObject(String table, DPSearchStatement criteria,
            DPDocumentMapper<T> mapper) {
        Map<String, Object> source = queryForMap(table, criteria, null, null);
        return mapper.mapper(source);
    }

    @Override
    public <T> T queryForObject(String table, DPSearchStatement criteria, Set<String> fields,
            DPDocumentMapper<T> mapper) {
        Map<String, Object> source = queryForMap(table, criteria, fields, null);
        return mapper.mapper(source);
    }

    @Override
    public <T> T queryForObject(String table, long id, DPDocumentMapper<T> mapper) {
        Map<String, Object> source = queryForObject(table, String.valueOf(id), null, null);
        return mapper.mapper(source);
    }

    @Override
    public <T> T queryForObject(String table, long id, Set<String> fields,
            DPDocumentMapper<T> mapper) {
        Map<String, Object> source = queryForObject(table, String.valueOf(id), fields, null);
        return mapper.mapper(source);
    }

    @Override
    public <T> T queryForObject(String table, String id, DPDocumentMapper<T> mapper) {
        Map<String, Object> source = queryForObject(table, id, null, null);
        return mapper.mapper(source);
    }

    @Override
    public <T> T queryForObject(String table, String id, Set<String> fields,
            DPDocumentMapper<T> mapper) {
        Map<String, Object> source = queryForObject(table, String.valueOf(id), fields, null);
        return mapper.mapper(source);
    }

    public void setClient(TransportClient client) {
        this.client = client;
    }

    @Override
    public int update(String table, DPSearchStatement criteria, Map<String, Object> newSources) {
        return update(table, criteria, newSources, false);
    }

    @Override
    public int update(String table, DPSearchStatement criteria,
            Map<String, Object> newSources, boolean upsert) {
        checkDBAndTable(table);
        
        SearchRequestBuilder requestBuilder = client.prepareSearch(table).setTypes(table)
                .setScroll(new TimeValue(SCROLL_TIME_OUT));
        if (criteria != null) {
            requestBuilder.setQuery(buildElasticsearchQuery(criteria));
        }
        SearchResponse scrollResp = requestBuilder.setSize(BATCH_DELETE_PER_SIZE).execute().actionGet();
        int effectedCount = 0; // 影响的记录个数
        while (true) {
            SearchHit[] searchHits = scrollResp.getHits().getHits();
            if (searchHits.length == 0) {
                break;
            }
            BulkRequestBuilder bulkRequestBuilder = client.prepareBulk();
            for (SearchHit searchHit : searchHits) {
                bulkRequestBuilder.add(new UpdateRequest(table, table, searchHit.getId()).doc(newSources).upsert(newSources));
            }
            BulkItemResponse[] responses = bulkRequestBuilder.execute().actionGet().getItems();
            for (BulkItemResponse item : responses) {
                effectedCount += item.<UpdateResponse>getResponse().getResult() == Result.UPDATED ? 1 : 0;
            }
            String scrollId = scrollResp.getScrollId();
            scrollResp = client.prepareSearchScroll(scrollId).setScroll(new TimeValue(SCROLL_TIME_OUT)).execute().actionGet();
        }
        return effectedCount;
    }

    @Override
    public int update(String table, long id, Map<String, Object> newSources) {
        client.prepareUpdate(table, table, String.valueOf(id)).setDoc(newSources);
        return 1;
    }
    
    @Override
    public int update(String table, long id, Map<String, Object> newSources, boolean upsert) {
        return update(table, String.valueOf(id), newSources, upsert);
    }

    
    @Override
    public int update(String table, String id, Map<String, Object> newSources) {
        client.prepareUpdate(table, table, id).setDoc(newSources);
        return 1;
    }

    
    @Override
    public int update(String table, String id, Map<String, Object> newSources, boolean upsert) {
        UpdateRequestBuilder requestBuilder = client.prepareUpdate(table, table, id).setDoc(newSources);
        if (upsert) {
            requestBuilder.setUpsert(newSources);
        }
        return requestBuilder.execute().actionGet().getResult() == Result.UPDATED ? 1 : 0;
    }
    
}
